package com.tenthoundsqps.service.sync;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.tenthoundsqps.dao.OrderDao;
import com.tenthoundsqps.entity.Order;
import com.tenthoundsqps.service.sync.dto.OrderSyncMessage;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;

/**
 * 订单同步Kafka消息监听器
 * 负责消费Kafka中的订单同步消息并执行实际的ES同步操作
 * 
 * 消费策略：
 * 1. 并发消费：支持多个消费者实例并行处理消息
 * 2. 容错处理：消息处理失败时记录日志并重新入队
 * 3. 幂等性：保证同一条消息多次消费结果一致
 */
@Service
public class OrderSyncKafkaListener {
    
    private static final Logger logger = LoggerFactory.getLogger(OrderSyncKafkaListener.class);
    
    /**
     * JSON对象映射器
     * 用于消息的序列化和反序列化
     */
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    /**
     * 订单数据访问对象
     * 用于从数据库查询订单详情
     */
    @Autowired
    private OrderDao orderDao;
    
    /**
     * Elasticsearch高级客户端
     * 用于执行索引和删除操作
     * 
     * 注意：RestHighLevelClient 在 Elasticsearch 7.15+ 中已被弃用，
     * 但在当前版本中仍可使用。建议在升级到 Elasticsearch 8.0+ 时
     * 迁移到新的 Java API Client。
     */
    @Autowired
    private RestHighLevelClient elasticsearchClient;
    
    /**
     * 订单同步服务
     * 用于复用同步逻辑
     */
    @Autowired
    private OrderSyncService orderSyncService;
    
    // Elasticsearch索引名称
    private static final String ORDER_INDEX = "orders";
    
    // 日期时间格式化器
    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    
    /**
     * 监听订单同步消息
     * 当有订单创建或更新时，消费此主题中的消息并同步到ES
     * 
     * @param message 消息内容
     */
    @KafkaListener(topics = "order-sync-topic", groupId = "order-sync-group")
    public void listenOrderSyncMessage(String message) {
        try {
            logger.info("Received order sync message: {}", message);
            
            // 使用JSON反序列化解析消息内容
            OrderSyncMessage syncMessage = objectMapper.readValue(message, OrderSyncMessage.class);
            
            // 判断操作类型
            if ("SYNC".equals(syncMessage.getAction())) {
                // 从数据库查询订单详情
                Order order = orderDao.selectById(syncMessage.getOrderId());
                if (order != null) {
                    // 同步到Elasticsearch
                    syncOrderToElasticsearch(order);
                    logger.info("Order sync completed, orderId: {}", syncMessage.getOrderId());
                } else {
                    logger.warn("Order not found in database, orderId: {}", syncMessage.getOrderId());
                }
            }
        } catch (Exception e) {
            logger.error("Failed to process order sync message: {}", message, e);
            // 实际应用中应将失败消息发送到死信队列或重试队列
        }
    }
    
    /**
     * 监听订单删除消息
     * 当有订单删除时，消费此主题中的消息并从ES中删除
     * 
     * @param message 消息内容
     */
    @KafkaListener(topics = "order-delete-topic", groupId = "order-sync-group")
    public void listenOrderDeleteMessage(String message) {
        try {
            logger.info("Received order delete message: {}", message);
            
            // 使用JSON反序列化解析消息内容
            OrderSyncMessage deleteMessage = objectMapper.readValue(message, OrderSyncMessage.class);
            
            // 判断操作类型
            if ("DELETE".equals(deleteMessage.getAction())) {
                // 从Elasticsearch中删除订单
                deleteOrderFromElasticsearch(deleteMessage.getOrderId());
                logger.info("Order delete completed, orderId: {}", deleteMessage.getOrderId());
            }
        } catch (Exception e) {
            logger.error("Failed to process order delete message: {}", message, e);
            // 实际应用中应将失败消息发送到死信队列或重试队列
        }
    }
    
    /**
     * 同步订单到Elasticsearch
     * 
     * @param order 订单对象
     */
    private void syncOrderToElasticsearch(Order order) {
        try {
            // 构造要同步到ES的文档数据
            Map<String, Object> document = buildOrderDocument(order);
            
            // 创建索引请求
            IndexRequest indexRequest = new IndexRequest(ORDER_INDEX);
            indexRequest.id(String.valueOf(order.getId())); // 使用订单ID作为文档ID
            indexRequest.source(document);
            
            // 执行索引操作
            elasticsearchClient.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("Failed to sync order to Elasticsearch, orderId: {}", order.getId(), e);
        }
    }
    
    /**
     * 从Elasticsearch中删除订单
     * 
     * @param orderId 订单ID
     */
    private void deleteOrderFromElasticsearch(Long orderId) {
        try {
            // 创建删除请求
            DeleteRequest deleteRequest = new DeleteRequest(ORDER_INDEX, String.valueOf(orderId));
            
            // 执行删除操作
            elasticsearchClient.delete(deleteRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("Failed to delete order from Elasticsearch, orderId: {}", orderId, e);
        }
    }
    
    /**
     * 构造订单文档数据
     * 将Order实体对象转换为Elasticsearch文档格式
     * 
     * @param order 订单对象
     * @return 文档数据Map
     */
    private Map<String, Object> buildOrderDocument(Order order) {
        Map<String, Object> document = new HashMap<>();
        
        // 映射数据库字段到Elasticsearch文档字段
        document.put("orderNo", order.getOrderNo());           // 订单编号
        document.put("userId", order.getUserId());             // 用户ID
        document.put("merchantId", order.getMerchantId());     // 商户ID
        document.put("merchantName", order.getMerchantName()); // 商户名称
        document.put("phoneNumber", order.getPhoneNumber());   // 联系电话
        document.put("amount", order.getAmount());             // 订单金额
        document.put("status", order.getStatus());             // 订单状态
        document.put("archived", order.getArchived());         // 归档状态
        
        // 时间字段需要特殊处理
        if (order.getCreateTime() != null) {
            document.put("createTime", order.getCreateTime().format(DATE_TIME_FORMATTER));
        }
        
        if (order.getUpdateTime() != null) {
            document.put("updateTime", order.getUpdateTime().format(DATE_TIME_FORMATTER));
        }
        
        return document;
    }
}